package com.tledu.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.util.*;

public class PracticeQuestions4 {
    public static void main(String[] args) throws IOException {
        // 1. 获取到对应文件夹下的所有文件，简单考虑的时候，直接拿文件夹下的文件，更完善一些，可能会存在文件夹下也有文件的可能性，需要递归获取到所有
        // 的文件
        //  1.1 输入一个文件夹
        String dirPath = "C:\\Users\\cyrus\\Desktop\\毕设公开课\\hdfs-job021\\data\\data";
        //  1.2 加载里面的所有文件地址
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.newInstance(conf);
        Path path = new Path(dirPath);
        // 获取文件夹下的文件了
        FileStatus[] statuses = fs.listStatus(path);
        List<String> pathList = new ArrayList<>();
        for (FileStatus status : statuses) {
            pathList.add(status.getPath().toString());
        }
        // 2. 读取这些文件，获取所有单词的数组
        List<String> wordList = new ArrayList<>();
        for (String filePath : pathList) {
            String result = HdfsUtils.readHdfsFile(filePath, "utf8");
            wordList.addAll(Arrays.asList(result.split(" ")));
        }

        // 3. 针对这些数据，进行词频统计，kv形式存储，k是单词，v就是频率
        Map<String, Integer> wordFreq = new HashMap<>();
        for (String word : wordList) {
            if (wordFreq.containsKey(word)) {
                wordFreq.put(word, wordFreq.get(word) + 1);
            }else {
                wordFreq.put(word, 1);
            }
        }

//        wordFreq.forEach((k, v) -> {
//            System.out.println("key:"+k+",value:"+v);
//        });

        // 4. 进行排序
        // 将map转成list，之后可以通过冒泡等方式进行排序
        // 或者这里，我们也可以通过对象数组的方式进行排序
        List<WordFreq> wordFreqList = new ArrayList<>();
        wordFreq.keySet().forEach(key->{
            wordFreqList.add(new WordFreq(key, wordFreq.get(key)));
        });
        wordFreqList.sort(new Comparator<WordFreq>() {
            @Override
            public int compare(WordFreq o1, WordFreq o2) {
                return o2.getFreq() - o1.getFreq();
            }
        });

        wordFreqList.forEach(item-> System.out.println(item.toString()));
    }
}
